package org.example.flink.operation;

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

public class Hello19ProcessFunctionSideOutPut {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> source = environment.fromElements("a", "aa", "aaa", "b", "bb", "bbb");
        //读取一次，数据全部取出
        OutputTag<String> outputTag1 = new OutputTag<>("outputTag1"){

        };
        OutputTag<String> outputTag2 = new OutputTag<>("outputTag1"){

        };
        SingleOutputStreamOperator<String> stream = source.process(new ProcessFunction<String, String>() {
            @Override
            public void processElement(String value, ProcessFunction<String, String>.Context context, Collector<String> collector)
                    throws Exception {
                if (value.length() == 1) {
                    context.output(outputTag1, value);
                } else if (value.length() == 2) {
                    context.output(outputTag2, value);
                } else {
                    collector.collect(value.toUpperCase());
                }
            }
        });

        //获取侧输出的数据
        stream.getSideOutput(outputTag1).print("侧输出1： ").setParallelism(1);
        stream.getSideOutput(outputTag2).print("侧输出2： ").setParallelism(1);

        //输出数据
        stream.print();

        environment.execute();
    }
}
